package cn.hy.dataSync.service;


import cn.hy.dataSync.alter.AlertManager;
import cn.hy.dataSync.bean.DataSourceConfig;
import cn.hy.dataSync.config.*;
import cn.hy.dataSync.job.JobContainer;
import cn.hy.dataSync.utils.TraceIdUtil;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 迁移服务
 *
 */
@Slf4j
@Service
public class MigrateService {

    /**
     * 迁移常常是一个花费时间的活儿，而请求线程往往又有超时控制，所以单独线程执行
     * <p>
     * 任务的先后顺序我认为不是一个要考量的事，如果一定要考量请用公平线程池
     */
    private static final ExecutorService EXECUTOR =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    /**
     * 默认速度限制
     */
    private static final SpeedConfig SPEED = SpeedConfig.builder()
            .channel(2)
            .record(5000L)
            .build();

    /**
     * 默认错误限制
     */
    private static final ErrorLimit ERROR_LIMIT = ErrorLimit.builder()
            .record(10L)
            .percentage(0.01D)
            .build();

    //@Resource
    //private DataSourceConfig dataSourceConfig;

    public Future<?> submitAsync(final JobConfig job) {
        return EXECUTOR.submit(() -> submit(job));
    }

    public void submit(final JobConfig job) {
        try {
            init(job);
        } catch (Exception ex) {
            log.error("迁移任务初始化失败，原因：{}", ex.getMessage());
            return;
        }

        log.info("迁移任务：{}", JSON.toJSONString(job));

        String traceId = TraceIdUtil.getCurrentTraceId();
        try {
            //用任务 ID 重置 traceId
            TraceIdUtil.initTraceId(job.getJobId());

            new JobContainer(job).start();
        } catch (Exception ex) {
            log.error("迁移任务 {} 提交失败，原因：{}", job.getJobId(), ex.getMessage());
            // AlertManager.sendText(String.format("迁移任务 %s 执行失败，traceId：%s\n", job.getJobId(), TraceIdUtil.getCurrentTraceId()) + JSON.toJSONString(job));
        } finally {
            //还原 traceId
            TraceIdUtil.initTraceId(traceId);
        }
    }

    private void init(final JobConfig job) {
        Validate.notNull(job, "job 不能为空");
        Validate.notNull(job.getReaderConfig(), "readerConfig 不能为空");
        Validate.notNull(job.getWriterConfig(), "writerConfig 不能为空");

        //数据库信息不存在时，默认使用当前库信息
    /*  ReaderConfig readerConfig = job.getReaderConfig();
        if (StringUtils.isBlank(readerConfig.getJdbcUrl())) {
            readerConfig.setJdbcUrl(dataSourceConfig.getUrl());
        }

        if (StringUtils.isBlank(readerConfig.getUsername())) {
            readerConfig.setUsername(dataSourceConfig.getUsername());
        }

        if (StringUtils.isBlank(readerConfig.getPassword())) {
            readerConfig.setPassword(dataSourceConfig.getPassword());
        }

        WriterConfig writerConfig = job.getWriterConfig();
        if (StringUtils.isBlank(writerConfig.getJdbcUrl())) {
            writerConfig.setJdbcUrl(dataSourceConfig.getUrl());
        }

        if (StringUtils.isBlank(writerConfig.getUsername())) {
            writerConfig.setUsername(dataSourceConfig.getUsername());
        }

        if (StringUtils.isBlank(writerConfig.getPassword())) {
            writerConfig.setPassword(dataSourceConfig.getPassword());
        }*/

        if (Objects.isNull(job.getSpeedConfig())) {
            job.setSpeedConfig(SPEED);
        }

        if (Objects.isNull(job.getErrorLimit())) {
            job.setErrorLimit(ERROR_LIMIT);
        }
    }
}
